﻿using Core.Entity.Commer.Entity;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Core.Commer.MQ.Connection
{
    public class RabbitMQConnection
    {
        private readonly MQConfiguration _options;
        private readonly GetMessageInfo _messageInfo;

        public RabbitMQConnection(IOptions<MQConfiguration> options, GetMessageInfo messageInfo)
        {
            _options = options.Value;
            _messageInfo = messageInfo;
        }

        #region 普通模式
        /// <summary>
        /// 普通模式 生产者
        /// </summary>
        /// <param name="mQ">相关配置信息</param>
        /// <returns></returns>
        public async Task<bool> RMQSendInfoMessege(MQConfiguration mQ)
        {
            if (!mQ.message.Any()) return false;
            try
            {
                IConnectionFactory connFactory = new ConnectionFactory
                {
                    HostName = _options.hosturl,
                    Port = _options.port,
                    UserName = _options.username,
                    Password = _options.password
                };
                //创建连接对象
                using (IConnection connection = connFactory.CreateConnection())
                {
                    //创建连接会话对象
                    using (IModel channel = connection.CreateModel())
                    {
                        //声明一个队列
                        // durable:true 消息持久化
                        channel.QueueDeclare(queue: mQ.queuename, durable: false, exclusive: false, autoDelete: false, arguments: null);

                        //发送消息
                        channel.BasicPublish(exchange: "", routingKey: mQ.queuename, basicProperties: null, body: Encoding.UTF8.GetBytes(_messageInfo.GetMessage(mQ.message)));
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("消息发送失败，失败原因=>" + ex);
                return false;
            }

            return true;
        }


        /// <summary>
        /// 普通模式 消费者
        /// </summary>
        /// <param name="mQ">相关配置信息</param>
        public string MqReceivingInfoMessage(MQConfiguration mQ)
        {
            string message = string.Empty;
            try
            {
                IConnectionFactory connFactory = new ConnectionFactory
                {
                    HostName = _options.hosturl,
                    Port = _options.port,
                    UserName = _options.username,
                    Password = _options.password
                };
                using (IConnection connection = connFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        //durable:true  消息持久化
                        channel.QueueDeclare(queue: mQ.queuename, durable: false, exclusive: false, autoDelete: false, arguments: null);

                        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

                        consumer.Received += (model, ea) =>
                        {
                            string messages = Encoding.UTF8.GetString(ea.Body.ToArray());
                        };
                        channel.BasicConsume(queue: mQ.queuename, autoAck: true, consumer: consumer);
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("消息消费失败，失败原因=>" + ex);
                return string.Empty;
            }
            return message;
        }
        #endregion


        #region 工作模式 Work
        /// <summary>
        /// 生产者
        /// </summary>
        /// <param name="mQ">MQ 配置类</param>
        /// <returns></returns>
        public async Task<bool> WorkeSendMessege(MQConfiguration mQ)
        {
            if (!mQ.message.Any()) return false;
            try
            {
                IConnectionFactory connFactory = new ConnectionFactory
                {
                    HostName = _options.hosturl,
                    Port = _options.port,
                    UserName = _options.username,
                    Password = _options.password
                };
                //创建连接对象
                using (IConnection connection = connFactory.CreateConnection())
                {
                    //创建连接会话对象
                    using (IModel channel = connection.CreateModel())
                    {
                        //声明一个队列
                        // durable:true 消息持久化
                        channel.QueueDeclare(queue: mQ.queuename,
                            durable: true,
                            exclusive: false,
                            autoDelete: false,
                            arguments: null);

                        string message = _messageInfo.GetMessage(mQ.message.ToArray());

                        // 设置为持久性
                        IBasicProperties properties = channel.CreateBasicProperties();
                        properties.Persistent = true;

                        // 平均分配，在工作者还没处理完消息，并返回确认标志钱，不要给它新的消息
                        channel.BasicQos(0, 1, false);

                        int dots = message.Split(".").Length - 1;
                        Thread.Sleep(dots * 1000);

                        channel.BasicPublish(exchange: "",
                                 routingKey: mQ.queuename,
                                 basicProperties: properties,
                                 body: Encoding.UTF8.GetBytes(message));
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("消息发送失败，失败原因=>" + ex);
                return false;
            }

            return true;
        }

        /// <summary>
        /// 工作模式，消费消息
        /// </summary>
        /// <param name="mQ"></param>
        public string WorkeReceivingMessage(MQConfiguration mQ)
        {
            string message = string.Empty;
            IConnectionFactory connFactory = new ConnectionFactory
            {
                HostName = _options.hosturl,
                Port = _options.port,
                UserName = _options.username,
                Password = _options.password
            };
            try
            {
                using (IConnection connection = connFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        //durable:true  消息持久化
                        channel.QueueDeclare
                            (
                            queue: mQ.queuename,
                            durable: true,
                            exclusive: false,
                            autoDelete: false,
                            arguments: null
                            );

                        channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

                        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

                        consumer.Received += (model, ea) =>
                        {
                            message = Encoding.UTF8.GetString(ea.Body.ToArray());

                            Console.WriteLine("工作模式收到消息=》" + message);

                            int dots = message.Split(".").Length - 1;

                            Thread.Sleep(dots * 1000);

                            // 进行消息确认
                            channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);

                        };
                        //消息确认模式，分为：自动确认模式和手动确认模式

                        // 自动确认模式（autoAck：true ）
                        //channel.BasicConsume(queue: mQ.queuename, autoAck: true, consumer: consumer);

                        //手动确认模式（autoAck ：false）
                        channel.BasicConsume(queue: mQ.queuename, autoAck: false, consumer: consumer);

                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("工作模式消息消费失败=》" + ex);
                return string.Empty;
            }
            return message;
        }
        #endregion


        #region 发布订阅模式 Publish/Subscribe
        /// <summary>
        /// 发布订阅模式 生产者
        /// </summary>
        /// <param name="mQ">MQ 配置类</param>
        /// <returns></returns>
        public async Task<bool> PublishSubscribeSendMessege(MQConfiguration mQ)
        {
            if (!mQ.message.Any()) return false;
            try
            {
                IConnectionFactory connFactory = new ConnectionFactory
                {
                    HostName = _options.hosturl,
                    Port = _options.port,
                    UserName = _options.username,
                    Password = _options.password
                };
                //创建连接对象
                using (IConnection connection = connFactory.CreateConnection())
                {
                    //创建连接会话对象
                    using (IModel channel = connection.CreateModel())
                    {
                        // 声明交换机，并指定类型

                        channel.ExchangeDeclare(exchange: mQ.exchangename, type: ExchangeType.Fanout);

                        //获取消息
                        string seedmessage = _messageInfo.GetMessage(mQ.message);

                        //发送消息
                        channel.BasicPublish(exchange: mQ.exchangename,
                                                routingKey: "",
                                                basicProperties: null,
                                                body: Encoding.UTF8.GetBytes(seedmessage)
                                              );

                        //Console.WriteLine("发布订阅模式消息发送成功=》" + seedmessage);

                        Console.WriteLine(" [x] Sent {0}", seedmessage);
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("消息发送失败，失败原因=>" + ex);
                return false;
            }

            Console.WriteLine(" Press [enter] to exit.");

            Console.ReadLine();

            return true;
        }

        /// <summary>
        /// 发布订阅模式，消费消息
        /// </summary>
        /// <param name="mQ"></param>
        public void PublishSubscribeReceivingMessage(MQConfiguration mQ)
        {
            IConnectionFactory connFactory = new ConnectionFactory
            {
                HostName = _options.hosturl,
                Port = _options.port,
                UserName = _options.username,
                Password = _options.password
            };
            using (IConnection connection = connFactory.CreateConnection())
            {
                using (IModel channel = connection.CreateModel())
                {
                    // 匹配 发布 订阅
                    channel.ExchangeDeclare(exchange: mQ.exchangename, type: ExchangeType.Fanout);

                    //声明队列：使用生成的名称创建一个非持久的，独占的自动删除
                    string queueName = channel.QueueDeclare().QueueName;

                    //绑定队列
                    channel.QueueBind(queue: queueName, exchange: mQ.exchangename, routingKey: "");

                    EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

                    //消费时触发该事件
                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());

                        Console.WriteLine(" [x] {0}", message);
                    };

                    //消费消息
                    channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);

                    Console.WriteLine("按[enter]退出");

                    Console.ReadLine();
                }
            }
        }
        #endregion


        #region 路由模式 Routing

        /// <summary>
        /// 生产者发送消息
        /// </summary>
        /// <param name="mQ"></param>
        /// <returns></returns>
        public bool MQRoutingSenMessage(MQConfiguration mQ)
        {
            if (!mQ.message.Any()) return false;
            try
            {
                IConnectionFactory connFactory = new ConnectionFactory
                {
                    HostName = _options.hosturl,
                    Port = _options.port,
                    UserName = _options.username,
                    Password = _options.password
                };
                //创建连接对象
                using (IConnection connection = connFactory.CreateConnection())
                {
                    //创建连接会话对象
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchange: mQ.exchangename, type: ExchangeType.Direct);

                        string message = _messageInfo.GetMessage(mQ.message);

                        channel.BasicPublish(exchange: mQ.exchangename,
                                             routingKey: mQ.routingKey,
                                             basicProperties: null,
                                             body: Encoding.UTF8.GetBytes(message));

                        Console.WriteLine("路由模式数据创建成功=》" + message);
                    }

                    Console.WriteLine(" Press [enter] to exit.");

                    Console.ReadLine();
                }
            }
            catch (Exception ex)
            {
                return false;
            }
            return true;
        }


        /// <summary>
        /// 消费者消费消息
        /// </summary>
        /// <param name="mQ"></param>
        public void MQRoutingReceivingMessage(MQConfiguration mQ)
        {
            IConnectionFactory connFactory = new ConnectionFactory
            {
                HostName = _options.hosturl,
                Port = _options.port,
                UserName = _options.username,
                Password = _options.password
            };

            using (var connection = connFactory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare(exchange: mQ.exchangename, type: ExchangeType.Direct);

                    var queueName = channel.QueueDeclare().QueueName;

                    if (mQ.message != null && mQ.message.Length < 1)
                    {
                        Console.Error.WriteLine("Usage: {0} [info] [warning] [error]",
                                                Environment.GetCommandLineArgs()[0]);
                        Console.WriteLine(" Press [enter] to exit.");
                        Console.ReadLine();
                        Environment.ExitCode = 1;
                        return;
                    }

                    foreach (string severity in mQ.message)
                    {
                        channel.QueueBind(queue: queueName,
                                          exchange: mQ.exchangename,
                                          routingKey: mQ.routingKey);
                    }

                    Console.WriteLine(" [*] Waiting for messages.");

                    EventingBasicConsumer consumer = new EventingBasicConsumer(channel);

                    consumer.Received += (model, ea) =>
                    {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());

                        var routingKey = ea.RoutingKey;

                        Console.WriteLine(" [x] Received '{0}':'{1}'",
                                          routingKey, message);

                    };
                    channel.BasicConsume(queue: queueName,
                                         autoAck: true,
                                         consumer: consumer);

                    Console.WriteLine(" Press [enter] to exit.");
                    Console.ReadLine();
                }
            }
        }

        #endregion

        #region 通配符模式
        /// <summary>
        /// 通配符模式：生产者
        /// </summary>
        /// <param name="mQ"></param>
        /// <returns></returns>
        public bool MQWildcardSenMessage(MQConfiguration mQ)
        {
            if (!mQ.message.Any()) return false;
            try
            {
                IConnectionFactory connFactory = new ConnectionFactory
                {
                    HostName = _options.hosturl,
                    Port = _options.port,
                    UserName = _options.username,
                    Password = _options.password
                };
                using (IConnection connection = connFactory.CreateConnection())
                {
                    using (IModel channel = connection.CreateModel())
                    {
                        channel.ExchangeDeclare(exchange: mQ.exchangename, type: "topic");

                        //var routingKey = (mQ.message.Length > 0) ? mQ.message[0] : "anonymous.info";

                        var message = _messageInfo.GetMessage(mQ.message);

                        var body = Encoding.UTF8.GetBytes(message);

                        channel.BasicPublish(exchange: mQ.exchangename,
                                             routingKey: mQ.routingKey,
                                             basicProperties: null,
                                             body: body);

                        Console.WriteLine(" [x] Sent '{0}':'{1}'", mQ.routingKey, message);
                    }
                }
            }
            catch (Exception ex)
            {
                throw ex;
            }
            return true;
        }
        #endregion
    }
}
